package com.xzx.flink.streamapi.sink;

import com.xzx.flink.bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2021-08-29 10:41 下午
 */
public class Sink_04_ES {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = env.readTextFile("/Users/xinzhixuan/work/git/flink-study-java/src/main/resources/sensor.txt");
        SingleOutputStreamOperator<SensorReading> map = streamSource.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String value) throws Exception {
                String[] split = value.split(" ");
                return new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
            }
        });
        map.print();

        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));

        ElasticsearchSink.Builder<SensorReading> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts, new ElasticsearchSinkFunction<SensorReading>() {

            @Override
            public void open() throws Exception {
                ElasticsearchSinkFunction.super.open();
            }

            @Override
            public void close() throws Exception {
                ElasticsearchSinkFunction.super.close();
            }

            @Override
            public void process(SensorReading element, RuntimeContext ctx, RequestIndexer indexer) {
                indexer.add(createIndexRequest(element));
            }

            public IndexRequest createIndexRequest(SensorReading element) {
                Map<String, Object> json = new HashMap<>();
                json.put("id", element.getId());
                json.put("timestamp", element.getTimestamp());
                json.put("temperature", element.getTemperature());

                return Requests.indexRequest()
                        .index("sensor")
                        .id(element.getId())
                        .source(json);
            }

        });
        map.addSink(esSinkBuilder.build());

        env.execute(Sink_04_ES.class.getName());
    }
}
